package com.newegg.whois.service;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.newegg.whois.controller.SearchAPI;
import com.newegg.whois.util.ExecutorQueue;
import com.newegg.whois.util.IPHelp;
import com.newegg.whois.util.RestTemplateUtil;

@Component
public class ConsumerWorkerThread{
	private static final Logger logger = Logger.getLogger(ConsumerWorkerThread.class);
	KafkaConsumer<String, String> consumer;
	
	@Value("${kafka.enable}")
	boolean enable;
	
	@Value("${kafka.consumer.host}")
	String consumerHost;
	
	@Value("${kafka.consumer.group}")
	String group;
	
	@Value("${kafka.consumer.group.size}")
	int groupSize;
	
	@Value("${kafka.topic}")
	String topic;
	
	Thread task = null;
	boolean started = false;
	
	@Autowired
	SearchAPI search;
	
	@Value("${whois.search.url:}")
	String whois_url;
	
	@Autowired
	ElasticsearchTarget es;
	
	ExecutorQueue execute;
	
	@PostConstruct
	public void start() {
		if(!enable){ return; }
		execute = new ExecutorQueue(groupSize);
		Properties props = new Properties();
		props.put("bootstrap.servers", consumerHost);
		props.put("group.id", group);
		props.put("auto.offset.reset", "earliest");
		props.put("auto.commit.enable", "true");
		props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "10000");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("rebalance.backoff.ms", "2000");
		props.put("rebalance.max.retries", "5");
		props.put("max.poll.records", groupSize + "");
		this.consumer = new KafkaConsumer<String, String>(props);
		logger.info("Kafka consumer created successfully, configuration: {}" + props);
		started = true;
		consumer.subscribe(Arrays.asList(topic));
		this.task = new Thread(new Runnable() {
			@Override
			public void run() {
				while(started){
					try {
						ConsumerRecords<String, String> records = consumer.poll(10);
						for (final ConsumerRecord<String, String> record : records) {
							execute.submit(new Runnable() {
								@Override
								public void run() {
									try {
										execute(record);
									} catch (InterruptedException e) {
										logger.error("log error", e);
									}
								}
							});
						}
					} catch (Exception e) {
						logger.error("kafka error", e);
					}
				}
			}
		});
		this.task.start();
	}
	
	@SuppressWarnings("unchecked")
	public void execute(ConsumerRecord<String, String> record) throws InterruptedException {
		String data = record.value();
		List<Map<String, String>> recordLogs = new ArrayList<>();
		JSONObject value = JSONObject.parseObject(data);
		JSONArray logs = value.getJSONArray("Log");
		for (int i = 0; i < logs.size(); i++) {
			JSONObject log = logs.getJSONObject(i);
			for (String key : log.keySet()) {
				JSONObject itemObj = log.getJSONObject(key);
				String QN = itemObj.getString("QN");
				if(QN != null){
					QN = QN.replaceAll("\\([^)]*\\)", ".");
					if(QN.startsWith(".")){ QN = QN.substring(1); }
					if(QN.endsWith(".")){ QN = QN.substring(0, QN.length() - 1); }
				}
				final String domain = QN.trim();
				try {
					Map<String, String> item = itemObj.toJavaObject(Map.class);
					item.put("domain", domain);
					recordLogs.add(item);
				} catch (Exception e) {
					logger.error("onItem error", e);
				}
			}
		}
		if(recordLogs.size()>0) {
	        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(recordLogs.size());
	        List<Callable<Map<String, Object>>> calls = new ArrayList<>();
	        for (Map<String, String> map : recordLogs) {
				calls.add(new Callable<Map<String, Object>>() {
					@Override
					public Map<String, Object> call() throws Exception {
						String domain = map.get("domain");
						return onItem(domain, map);
					}
				});
			}
	        List<Future<Map<String, Object>>> allList = scheduledExecutorService.invokeAll(calls);
	        List<Map<String, Object>> items = new ArrayList<>();
			for (Future<Map<String, Object>> future : allList) {
				try {
					items.add(future.get());
				} catch (Exception e) {
					logger.error("query item error", e);
				}
			}
			try {
				es.importDatas(items);
			} catch (Exception e) {
				logger.error("send to es error", e);
			}
		}
	}
	
	public Map<String, Object> onItem(String domain, Map<String, String> map) {
		boolean intranet = (domain.toLowerCase().endsWith("corp"));
		boolean ipaddress = IPHelp.isIPAddress(domain);
		boolean isHostname = domain.indexOf(".") == -1;
		Map<String, Object> data = null;
		if(!ipaddress && !isHostname && !intranet){
			if(whois_url != null && !"".equals(whois_url)){
				try {
					data = RestTemplateUtil.getJsonMap(whois_url.replace("{domain}", domain));
					if(data != null && data.size() > 0){
						data.put("isDomain", true);
					}else{
						data = new HashMap<>();
					}
				} catch (Exception e) {
					data = new HashMap<>();
				}
			}else{
				try {
					data = search.query(domain, false);
					data.put("isDomain", true);
				} catch (Exception e) {
					data = new HashMap<>();
				}
			}
		}else{
			data = new HashMap<>();
		}
		data.put("SITE", domain);
		data.put("isIP", ipaddress);
		data.put("isHostname", isHostname);
		data.put("intranet", intranet);
		data.putAll(map);
		return data;
	}
	
	@PreDestroy
	public void stop() {
		this.started = false;
		this.task.interrupt();
		consumer.close();
	}
}
